package net.wlm.jtp.dwd.log.function;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class LogSplitProcessFunction extends ProcessFunction<String, String> {
    /*
     * 侧输出流时标记OutputTag实例
     */
   private final OutputTag<String> errorTag;
   private final OutputTag<String> startTag;
   private final OutputTag<String> displayTag;
   private final OutputTag<String> actionTag; // 私有常量
   public LogSplitProcessFunction(OutputTag<String> errorTag,
                                  OutputTag<String> startTag,
                                  OutputTag<String> displayTag,
                                  OutputTag<String> actionTag) {
       this.errorTag = errorTag;
       this.startTag = startTag;
       this.displayTag = displayTag;
       this.actionTag = actionTag;
//       this.dateFormat = new SimpleDateFormat(DateTimeUtil.DATE_TIME_FORMAT);
   }
    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        // a.解析json数据
        JSONObject jsonObject = JSON.parseObject(value);
        // b.错误日志
        if(null !=  jsonObject.getString("err")){
            // 侧输出流输出
            ctx.output(errorTag, jsonObject.toJSONString());
            jsonObject.remove("err"); //  移除err字段
        }
        // c.启动日志
        if(null !=  jsonObject.getString("start")){
            // 侧输出流输出
            ctx.output(startTag,jsonObject.toJSONString());
        }
        // d.页面日志
        else {
            // 获取曝光和动作
            JSONArray displayJsonArray = jsonObject.getJSONArray("displays");
            JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");
            // 删除曝光和动作字段
            jsonObject.remove("displays");
            jsonObject.remove("actions");
            // 输出页面日志
            out.collect(jsonObject.toJSONString());
            if(null != displayJsonArray && !displayJsonArray.isEmpty()){
                for(int i = 0; i < displayJsonArray.size(); i++){
                    JSONObject displayJSONObject = displayJsonArray.getJSONObject(i);
                    // 添加字段
                    jsonObject.put("display", displayJSONObject.toJSONString());
                    // 输出
                    ctx.output(displayTag, jsonObject.toJSONString());
                }
            }
            if(null != actionsJsonArray && actionsJsonArray.isEmpty()){
                for (int i =0 ; i < actionsJsonArray.size(); i++ ){
                    JSONObject actionJsonObject = actionsJsonArray.getJSONObject(i);
                    // 添加字段
                    jsonObject.put("action", actionJsonObject.toJSONString());
                    // 输出
                    ctx.output(actionTag, jsonObject.toJSONString());
                }
            }
        }
    }
}